package cn._51doit.flink.day03.transformations;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class KeyByDemo7 {


    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //order01,辽宁省,沈阳市,8000
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        //对数据进行映射
        SingleOutputStreamOperator<Tuple4<String, String, String, Integer>> tpStream = lines.map(new MapFunction<String, Tuple4<String, String, String, Integer>>() {
            @Override
            public Tuple4<String, String, String, Integer> map(String line) throws Exception {
                String[] fields = line.split(",");
                String oid = fields[0];
                String province = fields[1];
                String city = fields[2];
                int money = Integer.parseInt(fields[3]);
                return Tuple4.of(oid, province, city, money);
            }
        });

        //按照下标进行keyBy
        //KeyedStream<Tuple4<String, String, String, Integer>, Tuple> keyedStream = tpStream.keyBy(1, 2);
        //KeyedStream<Tuple4<String, String, String, Integer>, Tuple> keyedStream = tpStream.keyBy("f1", "f2");

        //使用keySelect的匿名内部类
//        KeyedStream<Tuple4<String, String, String, Integer>, Tuple2<String, String>> keyedStream = tpStream.keyBy(new KeySelector<Tuple4<String, String, String, Integer>, Tuple2<String, String>>() {
//            @Override
//            public Tuple2<String, String> getKey(Tuple4<String, String, String, Integer> tp) throws Exception {
//                return Tuple2.of(tp.f1, tp.f2);
//            }
//        });

        //使用lambda表达式
        //KeyedStream<Tuple4<String, String, String, Integer>, Tuple2<String, String>> keyedStream = tpStream.keyBy(tp -> Tuple2.of(tp.f1, tp.f2), Types.TUPLE(Types.STRING, Types.STRING));

        KeyedStream<Tuple4<String, String, String, Integer>, String> keyedStream = tpStream.keyBy(tp -> tp.f1 + tp.f2);

        keyedStream.print();

        env.execute();




    }

}
